package com.atguigu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV1;
import com.atguigu.realtime.bean.TradeProvinceOrderWindow;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.function.DimAsyncFunction;
import com.atguigu.realtime.util.AtguiguUtil;
import com.atguigu.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @Author lzc
 * @Date 2022/7/29 13:53
 */
public class Dws_10_DwsTradeProvinceOrderWindow extends BaseAppV1 {
    public static void main(String[] args) {
        new Dws_10_DwsTradeProvinceOrderWindow().init(
            3010,
            2,
            "Dws_10_DwsTradeProvinceOrderWindow",
            Constant.TOPIC_ODS_DB
        );
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStream = stream
            .map(JSON::parseObject)
            
            .filter(obj ->
                        "gmall2022".equals(obj.getString("database"))
                            && "order_info".equals(obj.getString("table"))
                            && "insert".equals(obj.getString("type"))
            )
            .map(new MapFunction<JSONObject, TradeProvinceOrderWindow>() {
                @Override
                public TradeProvinceOrderWindow map(JSONObject value) throws Exception {
                    
                    JSONObject data = value.getJSONObject("data");
                    return TradeProvinceOrderWindow.builder()
                        .provinceId(data.getString("province_id"))
                        .orderCount(1L)
                        .orderAmount(data.getDoubleValue("original_total_amount"))
                        .ts(value.getLong("ts") * 1000)
                        .build();
                }
            })
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<TradeProvinceOrderWindow>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((bean, ts) -> bean.getTs())
                    .withIdleness(Duration.ofSeconds(5))
            )
            .keyBy(TradeProvinceOrderWindow::getProvinceId)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .reduce(
                new ReduceFunction<TradeProvinceOrderWindow>() {
                    @Override
                    public TradeProvinceOrderWindow reduce(TradeProvinceOrderWindow bean1,
                                                           TradeProvinceOrderWindow bean2) throws Exception {
                        
                        bean1.setOrderCount(bean1.getOrderCount() + bean2.getOrderCount());
                        bean1.setOrderAmount(bean1.getOrderAmount() + bean2.getOrderAmount());
                        return bean1;
                    }
                },
                new ProcessWindowFunction<TradeProvinceOrderWindow, TradeProvinceOrderWindow, String, TimeWindow>() {
                    @Override
                    public void process(String key,
                                        Context ctx,
                                        Iterable<TradeProvinceOrderWindow> elements,
                                        Collector<TradeProvinceOrderWindow> out) throws Exception {
                        TradeProvinceOrderWindow bean = elements.iterator().next();
                        
                        bean.setStt(AtguiguUtil.toDateTime(ctx.window().getStart()));
                        bean.setEdt(AtguiguUtil.toDateTime(ctx.window().getEnd()));
                        
                        bean.setTs(System.currentTimeMillis());
                        
                        out.collect(bean);
                    }
                }
            );
        
        AsyncDataStream
            .unorderedWait(
                beanStream,
                new DimAsyncFunction<TradeProvinceOrderWindow>() {
                    @Override
                    public String getTable() {
                        return "dim_base_province";
                    }
                    
                    @Override
                    public String getId(TradeProvinceOrderWindow input) {
                        return input.getProvinceId();
                    }
                    
                    @Override
                    public void addDim(TradeProvinceOrderWindow input, JSONObject dim) {
                        input.setProvinceName(dim.getString("NAME"));
                    }
                },
                60,
                TimeUnit.SECONDS
            )
            .addSink(FlinkSinkUtil.getClickHouseSink("dws_trade_province_order_window", TradeProvinceOrderWindow.class));
        
        
    }
}
